#include "cybertron/network/SocketTcpClient.hpp"
#include "cybertron/network/SocketEventHandler.hpp"
#include "cybertron/core/Log.hpp"
#include "AsioSocketAsync.hpp"
#include "SocketTcpClientThreads.hpp"
#include <asio.hpp>

CYBERTRON_BEGIN

class TcpClientBuiltinEventHandler : public SocketEventHandler
{
public:
	TcpClientBuiltinEventHandler(SocketTcpClient* pClient)
	{
		mpTcpClient = pClient;
	}

	virtual void onDisconnected()
	{
		if (mpTcpClient)
			mpTcpClient->notifyDisconnected();
	}

	virtual void onConnected()
	{
		if (mpTcpClient)
			mpTcpClient->notifyConnected();
	}

	virtual void onMessage(Message& msg)
	{
		if (mpTcpClient)
			mpTcpClient->notifyMessage(msg);
	}

private:
	SocketTcpClient* mpTcpClient = nullptr;
};

std::weak_ptr<SocketTcpClientThreads> gClientThreads;


CybertronCoreAPI SocketTcpClient::SocketTcpClient()
{
	mInOnTimer = false;
}

CybertronCoreAPI SocketTcpClient::~SocketTcpClient()
{
	closeThis();
	mFuncOnDisconnected = nullptr;
	mFuncOnConnected = nullptr;
	mFuncOnMessage = nullptr;
}
CybertronCoreAPI bool SocketTcpClient::resolverName(const std::string& ip, int port) {
	try
	{
			asio::ip::tcp::resolver::query addr(ip.c_str(), std::to_string(port).c_str());
			asio::io_context c;
			asio::ip::tcp::resolver r(c);
			r.resolve(addr);
	}
	catch (const std::exception& e)
	{
		logInfo("resolverName error %s", e.what());
		return false;
	}
	return true;
}
CybertronCoreAPI bool SocketTcpClient::connect(const std::string& ip, int port)
{
	if (mpContext != nullptr)
	{
		throw std::runtime_error("Cannot connect twice!");
	}
	//auto asio::ip::tcp::resolver::query xx;

	

	auto pClientThread = gClientThreads.lock();
	bool isNewThread = false;
	if (pClientThread == nullptr)
	{
		isNewThread = true;
		pClientThread = std::make_shared<SocketTcpClientThreads>("Client");
		pClientThread->setupContext();
	}
	
	std::shared_ptr<AsioSocketAsync> pSocketImpl = std::make_shared<AsioSocketAsync>(
		pClientThread->getContext(),
		ESocketStatus_Closed,
		ip,
		port);

	setImpl(pSocketImpl);
	
	std::shared_ptr<TcpClientBuiltinEventHandler> pBuiltinHandler = 
		std::make_shared<TcpClientBuiltinEventHandler>(this);
	addHandler(pBuiltinHandler);

	asio::ip::tcp::resolver resolver(*(pClientThread->getContext()));
	
	auto endpoints = resolver.resolve({ ip.c_str(), std::to_string(port).c_str() });
	pSocketImpl->connect(endpoints, [pSocketImpl]()
	{
		pSocketImpl->startReceiving();
	});

	if (isNewThread)
	{
		pClientThread->start(1);
		gClientThreads = pClientThread;
	}

	pClientThread->addSocket(shared_from_this());

	mpContext = pClientThread;

	return true;
}

CybertronCoreAPI bool SocketTcpClient::connectBlocked(const std::string& ip, int port)
{
	try
	{
		while (getStatus() != ESocketStatus_Connected)
		{
			logInfo("Blocked connecting to %s:%d ...", ip.c_str(), port);
			close("Connect failed, close and retry");
			while (!connect(ip, port))
			{
				std::this_thread::sleep_for(std::chrono::milliseconds(100));
			}

			while (getStatus() == ESocketStatus_Connecting)
			{
				std::this_thread::sleep_for(std::chrono::milliseconds(100));
			}
		}
		logInfo("Blocked connect to %s:%d successed", ip.c_str(), port);

		return true;
	}
	catch (std::exception& e)
	{
		logError("Exception on connect: %s", e.what());
		return false;
	}
	catch (...)
	{
		logError("Exception on connect");
		return false;
	}
}

CybertronCoreAPI void SocketTcpClient::close(const char* reason)
{
	// flush messages
	onTimer();

	SocketBase::close(reason);

	onTimer();

	closeThis();

	mFuncOnMessages.clear();
}

void SocketTcpClient::closeThis()
{
	std::shared_ptr<SocketTcpClientThreads> pThreads = std::static_pointer_cast<SocketTcpClientThreads>(mpContext);
	if (pThreads != nullptr)
	{
		pThreads->removeSocket(this);
	}

	mpContext = nullptr;
}

CybertronCoreAPI void SocketTcpClient::setHeartbeatSeconds(unsigned int seconds)
{
	std::shared_ptr<SocketTcpClientThreads> pThreads = std::static_pointer_cast<SocketTcpClientThreads>(mpContext);
	if (pThreads != nullptr)
	{
		pThreads->setHeartbeatSeconds(seconds);
	}
}

CybertronCoreAPI unsigned int SocketTcpClient::getHeartbeatSeconds() const
{
	std::shared_ptr<SocketTcpClientThreads> pThreads = std::static_pointer_cast<SocketTcpClientThreads>(mpContext);
	if (pThreads != nullptr)
	{
		return pThreads->getHeartbeatSeconds();
	}

	return 0;
}

CybertronCoreAPI void SocketTcpClient::setHandlerDisconnected(std::function<void()> funcOnDisconnected)
{
	mFuncOnDisconnected = funcOnDisconnected;
}

CybertronCoreAPI void SocketTcpClient::setHandlerConnected(std::function<void()> funcOnConnected)
{
	mFuncOnConnected = funcOnConnected;
}

CybertronCoreAPI void SocketTcpClient::setHandlerMessage(std::function<void(Message&)> funcOnMessage)
{
	mFuncOnMessage = funcOnMessage;
}

CybertronCoreAPI void SocketTcpClient::setHandlerMessage(
	std::uint16_t msgId,
	std::function<void(Message&)> funcOnMessage)
{
	mFuncOnMessages[msgId] = funcOnMessage;
}

CybertronCoreAPI void SocketTcpClient::onTimer(bool bSingleMsg)
{
	if (mInOnTimer)
	{
		return;
	}

	auto thisRef(shared_from_this());

	mInOnTimer = true;

	if (bSingleMsg) {
		Message msg;
		ESocketEvent eventType;
		bool bValid = false;
		{
			std::unique_lock<std::recursive_mutex> lock(mEventMutex);
			if (mEventMessages.size() > 0)
			{
				bValid = true;
				msg = mEventMessages.front();
				eventType = mEventTypes.front();
				mEventMessages.pop_front();
				mEventTypes.pop_front();
			}
		}
		if (bValid) {
			processOneMessage(eventType, msg);
		}
	}
	else {
		std::list <Message> eventMessages;
		std::list <ESocketEvent> eventTypes;
		{
			std::unique_lock<std::recursive_mutex> lock(mEventMutex);
			mEventMessages.swap(eventMessages);
			mEventTypes.swap(eventTypes);
		}
		auto itType = eventTypes.begin();
		auto itMsg = eventMessages.begin();

		for (; itType != eventTypes.end(); itType++, itMsg++)
		{
			processOneMessage(*itType, *itMsg);
		}
	}

	mInOnTimer = false;
}

void SocketTcpClient::processOneMessage(ESocketEvent eventType, Message& msg)
{
	switch (eventType)
	{
	case ESocketEvent_Disconnected:
	{
		if (mFuncOnDisconnected != nullptr)
		{
			mFuncOnDisconnected();
		}

		onDisconnected();
		break;
	}
	case ESocketEvent_Connected:
	{
		if (mFuncOnConnected != nullptr)
		{
			mFuncOnConnected();
		}

		onConnected();
		break;
	}
	case ESocketEvent_Message:
	{
		if (mFuncOnMessage != nullptr)
		{
			mFuncOnMessage(msg);
		}

		std::uint16_t msgId = msg.parseMsgId();
		auto it = mFuncOnMessages.find(msgId);
		if (it != mFuncOnMessages.end())
		{
			((*it).second)(msg);
		}

		onMessage(msg);
		break;
	}
	default:
		break;
	}
}

void SocketTcpClient::notifyDisconnected()
{
	std::unique_lock<std::recursive_mutex> lock(mEventMutex);
	mEventTypes.push_back(ESocketEvent_Disconnected);
	mEventMessages.resize(mEventMessages.size() + 1);
}

void SocketTcpClient::notifyConnected()
{
	std::unique_lock<std::recursive_mutex> lock(mEventMutex);
	mEventTypes.push_back(ESocketEvent_Connected);
	mEventMessages.resize(mEventMessages.size() + 1);
}

void SocketTcpClient::notifyMessage(Message& msg)
{
	std::unique_lock<std::recursive_mutex> lock(mEventMutex);
	mEventTypes.push_back(ESocketEvent_Message);
	mEventMessages.push_back(msg);
}

CybertronCoreAPI void SocketTcpClient::onDisconnected()
{
}

CybertronCoreAPI void SocketTcpClient::onConnected()
{
}

CybertronCoreAPI void SocketTcpClient::onMessage(Message& msg)
{
}

CYBERTRON_END
